package cn.itcast.flink.examples;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Author itcast
 * Date 2022/1/11 9:30
 * Desc TODO
 */
public class WordCountLambda {
    public static void main(String[] args) throws Exception {
        //获取流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取参数 并行度
        env.setParallelism(1);
        //读取socket数据源
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);

        //flatMap 每行数据 -> Tuple2<String,Integer>
        /*source.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            String[] words = value.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word,1));
            }
        });*/

        //void flatMap(T value, Collector<O> out)
        source.flatMap(
                (String value, Collector<Tuple2<String, Integer>> out) -> {
                    Arrays.stream(value.split(" ")).map(t -> Tuple2.of(t, 1)).forEach(out::collect);
                })
                .returns(Types.TUPLE(Types.STRING,Types.INT))
                //分组 聚合 打印
                .keyBy(t -> t.f0)
                .sum(1)
                .print();
        //执行流环境
        env.execute();
    }
}
